/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.nio.ByteBuffer;

import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.values.IntType;
import edu.ucsd.db.bassarisk.data.values.IntType.IntTools;
import edu.ucsd.db.bassarisk.data.values.IntType.IntegerValueCodec;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.common.IColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.loading.ColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.reading.IColumnIndexReader.DrawerInfo;

public class FixedLengthAscendingIntegerDrawerIndexReader implements
    IDrawerIndexReader<IntType> {

  protected static class Cursor {
    protected int cardIndex;
  }

  private static final int CARD_CAPACITY = 64;
  private static final int ENTRY_COUNT_LOCATION = 4;
  private static final int indexVersionNumber = 1;
  private static final int VERSION_NUM_LOCATION = 2;

  private final Cursor current;
  private int dataCardCount;
  private ByteBuffer drawerBuffer;
  private IColumnSegmentInfo<IntType> drawerInfo;
  private final IntegerValueCodec intManip;
  private int valueIndexStart;

  public FixedLengthAscendingIntegerDrawerIndexReader() {
    this.current = new Cursor();
    this.current.cardIndex = -1; // before the first card
    this.intManip = IntTools.INSTANCE.getGenericCodec();
  }

  // Used for testing
  public FixedLengthAscendingIntegerDrawerIndexReader(final ByteBuffer drawerBuffer,
                                                      final Cursor cursor,
                                                      final IColumnSegmentInfo<IntType> drawerInfo,
                                                      final int dataCardCount) {
    this.drawerBuffer = drawerBuffer;
    this.current = cursor;
    this.drawerInfo = drawerInfo;
    this.dataCardCount = dataCardCount;
    this.intManip = IntTools.INSTANCE.getGenericCodec();
  }

  @Override
  public void getCardData(final int cardIndex, final CardInfo<IntType> info) {
    assert cardIndex >= 0;
    assert cardIndex < this.dataCardCount;
    assert cardIndex >= this.current.cardIndex;

    long firstEntry =
        cardIndex * CARD_CAPACITY + this.drawerInfo.getFirstEntryIndex();

    this.current.cardIndex = cardIndex;
    info.reset(cardIndex, CARD_CAPACITY, firstEntry);
  }

  @Override
  public boolean getCardDataForEntryIndex(final int targetEntryIndex,
                                          final CardInfo<IntType> info) {
    long localIndex = targetEntryIndex - this.drawerInfo.getFirstEntryIndex();

    // require this to be a valid index entry for this drawer
    assert localIndex >= 0;
    if (localIndex >= (this.dataCardCount * CARD_CAPACITY)) { return false; }

    int targetCard = (int) Math.floor(localIndex / CARD_CAPACITY);
    this.getCardData(targetCard, info);
    return true;
  }

  @Override
  public boolean getCardDataForValue(final IValueReference<IntType> value,
                                     final CardInfo<IntType> info) {
    int targetValue = this.intManip.getIntegerValue(value);
    int i = Math.max(0, this.current.cardIndex);
    // faster with binary search
    for (; i <= this.dataCardCount; ++i) {
      int byteOffset = this.valueIndexStart + i * 4;
      int cardMinValue = this.drawerBuffer.getInt(byteOffset);
      if (cardMinValue > targetValue) {
        int targetCard = i - 1;
        if (targetCard >= this.current.cardIndex) {
          this.getCardData(targetCard, info);
          return true;
        }
      }
    }
    return false;
  }

  @Override
  public int getDataCardCount() {
    return this.dataCardCount;
  }

  @Override
  public int getDrawerIndexVersion() {
    return indexVersionNumber;
  }

  @Override
  public boolean getNextCardData(final CardInfo<IntType> info) {
    int nextCard = this.current.cardIndex + 1;
    if (nextCard < this.dataCardCount) {
      this.getCardData(nextCard, info);
      return true;
    } else {
      return false;
    }
  }

  @Override
  public CardInfo<IntType> makeCardInfoInstance() {
    return new CardInfo<IntType>(IntTools.INSTANCE);
  }

  @Override
  public void resetDrawer(final ByteBuffer drawerBuffer,
                          final ColumnSegmentInfo<IntType> drawerInfo) {
    DrawerInfo<IntType> di = (DrawerInfo<IntType>) drawerInfo;
    int headerPageIndex = di.getDrawerHeaderPageIndex();
    int headerEndOffset = (headerPageIndex + 1) * ColumnStore.BYTES_IN_CARD;
    this.drawerInfo = drawerInfo;
    this.drawerBuffer = drawerBuffer;

    assert indexVersionNumber == drawerBuffer.getShort(headerEndOffset
        - VERSION_NUM_LOCATION);

    int totalValues = drawerBuffer.getShort(headerEndOffset - ENTRY_COUNT_LOCATION);
    
    intManip.setIntegerValue(drawerBuffer.getInt(0), drawerInfo.getFirstValue());
    intManip.setIntegerValue(drawerBuffer.getInt(4 * (totalValues - 1)), drawerInfo.getLastValue());
    
    if (totalValues >= CARD_CAPACITY) {
      this.valueIndexStart = headerEndOffset - 2 * ColumnStore.BYTES_IN_CARD;
    } else {
      this.valueIndexStart = headerEndOffset - ColumnStore.BYTES_IN_CARD;
    }
    this.dataCardCount = (int) Math.floor((float) totalValues / CARD_CAPACITY);

    this.current.cardIndex = -1;
  }

}
